#include "KStreamAction.h"
#include "KSocket.h"
#include "KStorage.h"
#include "KTaskTimer.h"
#include "KTicker.h"
#include <algorithm>

KStreamAction::KStreamAction(KPageList *poolRecv, KPageList *poolSend, KTimerQueue *queue) :
    m_sending(false),
    m_recv(new KStorage(poolRecv)),
    m_send(new KStorage(poolSend ? poolSend : poolRecv)),
    m_life(NULL)
{
    if (queue) {
        m_timer.reset(new KTaskTimer(queue));
        m_timer->append(std::bind(&KStreamAction::_timer, this));
        m_ticker.reset(new KTicker);
    }
}

KStreamAction::~KStreamAction()
{
    assert(!m_timer || !m_timer->isAlive());
}

size_t KStreamAction::bytesToSend() const
{
    return m_send->bytesAvailable();
}

KPageList *KStreamAction::recvPool() const
{
    return m_recv->poolUsed();
}

KPageList *KStreamAction::sendPool() const
{
    return m_send->poolUsed();
}

bool KStreamAction::post(const void *data, size_t length, bool send)
{
    {
        KLocker lock(*this);
        if (!m_send->write(data, length)) {
            return false;
        } else if (!send || m_sending) {
            return true;
        } else {
            m_sending = true;
        }
    }
    return doSend();
}

void KStreamAction::onClosing(bool stopped)
{
    if (m_timer) {
        m_timer->stop();
    }
}

void KStreamAction::onReceived(KKey *&key, void *data, size_t length, bool more)
{
    bool ok = m_recv->write(data, length);
    process(m_recv, ok);
}

void KStreamAction::onSent(KKey *&key, void *data, size_t length)
{
    if (m_timer) {
        if (size_t speed = maxSendKBps()) {
            double delay = std::min<double>(maxSendDelayMs(),
                length / (speed * 1.024) - m_ticker->elapsed() * 1000.0 + 0.5);
            if (delay >= 1.0) {
                m_life = lifeBegin();
                if (m_timer->restart(DWORD(delay))) {
                    return;
                }
                lifeEnd(m_life);
            }
        }
    }
    doSend();
}

void KStreamAction::onSocketChanged()
{
    if (KSocket *sock = socket()) {
        assert(!KSocket::isUdpSocket(*sock));
    }
}

void KStreamAction::process(KStorage *storage, bool ok)
{
    KDebugMsg("Warning: virtual function KStreamAction::process() called");
}

void KStreamAction::_timer()
{
    KDef::Life life = m_life;
    doSend();
    lifeEnd(life);
}

bool KStreamAction::doSend()
{
    if (!bytesToSend()) {
        for (;;) {
            {
                KLocker lock(*this);
                if (bytesToSend() > 0) {
                    // New data has been posted
                    break;
                }
                m_sending = false;
            }
            onSendComplete();
            return true;
        }
    }
    if (m_ticker) {
        m_ticker->reset();
    }
    if (send(m_send, maxSendSize())) {
        return true;
    } else {
        KLocker lock(*this);
        m_sending = false;
        return false;
    }
}